package org.apache.dolphinscheduler.api.service.impl;

import com.alibaba.fastjson.JSONObject;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.NodeRelationService;
import org.apache.dolphinscheduler.api.service.OutputInfoService;
import org.apache.dolphinscheduler.dao.entity.MemoryAddress;
import org.apache.dolphinscheduler.dao.entity.NodeRelation;
import org.apache.dolphinscheduler.dao.entity.OutputInfo;
import org.apache.dolphinscheduler.dao.mapper.MemoryAddressMapper;
import org.apache.dolphinscheduler.dao.mapper.NodeRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.OutputInfoMapper;

import com.alibaba.fastjson.JSONArray;
import org.apache.dolphinscheduler.service.memoryaddress.MemoryAddressService;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

import static org.apache.dolphinscheduler.common.Constants.DATA_LIST;

@Service
public class OutputInfoServiceImpl extends BaseServiceImpl implements OutputInfoService {
    @Autowired
    OutputInfoMapper outputInfoMapper;

    @Autowired
    MemoryAddressMapper memoryAddressMapper;

    @Autowired
    NodeRelationMapper nodeRelationMapper;
    @Override
    public Map<String, Object> getColumnInfo(int nodeId,int point) {
        List<OutputInfo> outputInfo=outputInfoMapper.getInfo(point,nodeId,2);
        Map<String, Object> map=new HashMap<>();
        Map<String, String> result=new HashMap<>();
        if(null !=outputInfo && outputInfo.size()>0){
            String info=outputInfo.get(0).getInfo();
            info=info.substring(1,info.length()-1);
            info=info.replace(" ","");
            String [] contents=info.split(",");
            for (int i = 0; i < contents.length; i++) {
                String[] column=contents[i].split("=");
                result.put(column[0],column[1]);
            }
        }
        map.put(DATA_LIST, result);
        putMsg(map, Status.SUCCESS);
        return map;
    }

    /**
     * 防止查看输出有点慢 将部分数据存储到数据库做中转
     */
    @Override
    public void saveOutputInfo(Dataset<Row> dataset, int nodeId) {
        System.out.println("打印输出开始");
        //插入列信息
        StructType structType = dataset.schema();
        Map<String, String> map = new HashMap<>();
        for (int i = 0; i < structType.size(); i++) {
            StructField structField = structType.apply(i);
            map.put(structField.name(), structField.dataType().typeName());
        }
        OutputInfo outputInfo = new OutputInfo();
        outputInfo.setInfo(map.toString());
        outputInfo.setNodeId(nodeId);
        outputInfo.setType(2);
        outputInfo.setPoint("point1");
        outputInfo.setCreateTime(new Date());
        outputInfoMapper.insert(outputInfo);

        //插入表信息列名
        StringBuilder stringBuilder = new StringBuilder();
        String[] columns = dataset.columns();
        for (int i = 0; i < columns.length; i++) {
            stringBuilder.append(columns[i]);
            if (i != columns.length - 1) {
                stringBuilder.append(",,");
            }
        }
//         stringBuilder.append("]");
        OutputInfo outputInfo2 = new OutputInfo();
        outputInfo2.setInfo(stringBuilder.toString());
        outputInfo2.setNodeId(nodeId);
        outputInfo2.setType(1);
        outputInfo2.setPoint("point1");
        outputInfo2.setCreateTime(new Date());
        outputInfoMapper.insert(outputInfo2);


        //插入表信息
        List<Row> list = dataset.limit(100).collectAsList();
        System.out.println("打印输出开始" + list.size());
        for (int i = 0; i < list.size(); i++) {
            Row row = list.get(i);
            StringBuilder stringBuilder1 = new StringBuilder();
            for (int i1 = 0; i1 < row.size(); i1++) {
                stringBuilder1.append(row.get(i1));
                if (i1 != row.size() - 1) {
                    stringBuilder1.append(",,");
                }
            }
            OutputInfo outputInfo1 = new OutputInfo();
            outputInfo1.setInfo(stringBuilder1.toString());
            outputInfo1.setNodeId(nodeId);
            outputInfo1.setType(1);
            outputInfo1.setPoint("point1");
            outputInfo1.setCreateTime(new Date());
            outputInfoMapper.insert(outputInfo1);


        }
    }

    @Override
    public Map<String, Object> getList(int nodeId,int point) {
        List<OutputInfo> outputInfos=outputInfoMapper.getInfo(point,nodeId,1);
        Map<String, Object> map=new HashMap<>();
        List list=new ArrayList();
        for (int i = 0; i < outputInfos.size(); i++) {
            OutputInfo outputInfo=outputInfos.get(i);
            String info=outputInfo.getInfo();
            String [] array=  info.split(",,");
            list.add(array);
        }
        map.put(DATA_LIST, list);
        putMsg(map, Status.SUCCESS);
        return map;
    }

    @Override
    public Map<String, Object> getXYList(int nodeId, int point, String[] X, String[] Y){
        Map<String, Object> map=new HashMap<>();
        List<NodeRelation> list = nodeRelationMapper.selectNodeIdByNextNodeId(nodeId);
        int id=list.get(0).getNodeId();
        List<OutputInfo> outputInfos=outputInfoMapper.getInfo(point,id,1);
        String [] colums = outputInfos.get(0).getInfo().split(",,");
        JSONObject jsonObject = new JSONObject();
        com.alibaba.fastjson.JSONArray resultArray=new com.alibaba.fastjson.JSONArray();
        com.alibaba.fastjson.JSONArray colArray=new com.alibaba.fastjson.JSONArray();
        for(int x=0;x<X.length;x++){
            int indexX=printArray(colums,X[x]);
            for(int y=0;y<Y.length;y++){
                int indexY=printArray(colums,Y[y]);
                String[] colum=new String[2];
                colum[0]=X[x];
                colum[1]=Y[y];
                colArray.add(colum);
                com.alibaba.fastjson.JSONArray dataArray=new com.alibaba.fastjson.JSONArray();
                for(int i=1;i<outputInfos.size();i++){
                    String result[] =outputInfos.get(i).getInfo().split(",,");
                    String[] array = new String[2];
                    array[0]=result[indexX];
                    array[1]=result[indexY];
                    dataArray.add(array);
                }
                resultArray.add(dataArray);
            }
        }
        jsonObject.put("XY",colArray);
        jsonObject.put("xyData",resultArray);
        map.put(DATA_LIST, jsonObject);
        putMsg(map, Status.SUCCESS);
        return map;
    }

    @Override
    public void delByProcessInstanceId(int processInstanceId) {
         outputInfoMapper.delByProcessInstanceId(processInstanceId);
    }
    @Override
    public void delByProcessDefinitionId(int processDefinitionId) {
        outputInfoMapper.delByProcessDefinitionId(processDefinitionId);
    }

    //遍历数组
    public static int printArray(String[] array,String value){
        for(int i = 0;i<array.length;i++){
            if(array[i].equals(value)){
                return i;
            }
        }
        return -1;//当if条件不成立时，默认返回一个负数值-1
    }

    public static String[] objectToArray(Object o){
        String x=o.toString();
        String xstr=x.replaceAll("\"","");
        String xstr1=xstr.replaceAll("\\[","");
        String xstr2=xstr1.replaceAll("]","");
        String[] result=xstr2.split(",");
        return result;
    }

    public static void main(String[] args) {
        /*String aa="[Augsberger,,175.0,24.0,5.5,0.4,0.0,[175.0,24.0,5.5,0.4]]";
        aa=aa.substring(1,aa.length()-1);
        String [] contents=aa.split(",,");
        for (int i = 0; i < contents.length; i++) {
            System.out.println(contents[i]);
        }*/
        String ss="0.110300084686438";
        double a=Double.parseDouble(ss);
        System.out.println(a);
    }
}
